package com.swsc.ai.tasks;

import com.swsc.ai.config.SparkSqlConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @describe: 上市公司为主体的数据计算
 * @author: DuanCXin
 * @created: 2023-10-19 10:05
 */
public class CompanyEntityTask extends WideTableTempTask {
    @Override
    public void createView(SparkSession session, String... str) {
        long startTime = System.currentTimeMillis();
        Dataset<Row> stk_coconcept_df = SparkSqlConf.getDataBySql(session, "stk_coconcept", "updatestate <= 1", "id",
                "companycode", "conceptcode", "innercode", "updatetime", "indistate");
        stk_coconcept_df.createOrReplaceTempView("stk_coconcept");

        Dataset<Row> stk_companyarchives_df = SparkSqlConf.getDataBySql(session, "stk_companyarchives", "updatestate <= 1 and astockcode is not null", "id",
                "astockcode", "companycode", "provincecode", "officezipcode","inducsrccode","regcapital");
        stk_companyarchives_df.createOrReplaceTempView("stk_companyarchives");

        Dataset<Row> stk_conceptlist_df = SparkSqlConf.getDataBySql(session, "stk_conceptlist", "updatestate <= 1", "id",
                "classcode", "subclasscode", "conceptcode");
        stk_conceptlist_df.createOrReplaceTempView("stk_conceptlist");

        Dataset<Row> stk_monthquote_df = SparkSqlConf.getDataBySql(session, "stk_monthquote", "updatestate <= 1", "id",
                "companycode", "pettm", "pb", "tradingdate");
        stk_monthquote_df.createOrReplaceTempView("stk_monthquote");

        Dataset<Row> stk_performanceletters_df = SparkSqlConf.getDataBySql(session, "stk_performanceletters", "updatestate <= 1", "id",
                "operatingrevenue",
                "grossprofit",
                "totalprofit",
                "totalassets",
                "sewithoutmi",
                "netoperatecashflow",
                "roe",
                "oryoy",
                "grossprofityoy",
                "opyoy",
                "totalprofityoy",
                "totalassettoboy",
                "netocflowyoy",
                "enddate",
                "companycode");
        stk_performanceletters_df.createOrReplaceTempView("stk_performanceletters");

        Dataset<Row> stk_exgindustry_df = SparkSqlConf.getDataBySql(session, "stk_exgindustry", "updatestate <= 1", "firstindustrycode",
                "secondindustrycode", "thirdindustrycode", "fourthindustrycode", "companycode", "updatetime");
        stk_exgindustry_df.createOrReplaceTempView("stk_exgindustry");
        long endTime = System.currentTimeMillis();
        long totalTime = endTime - startTime;
        System.out.println("创建独立视图成功！运行时间： " + totalTime + " 毫秒");
    }

    @Override
    public void calculateResult(SparkSession session, String dt, String fileName) {
        Dataset<Row> sqlDF = session.sql("select scp.*, sco.conceptcode,sco.classcode,sco.subclasscode, sm.pettm, sm.pb, ceil(sp.operatingrevenue/10000) as operatingrevenue,\n" +
                "       ceil(sp.grossprofit/10000) as grossprofit, ceil(sp.totalprofit/10000) as totalprofit, ceil(sp.totalassets/10000) as totalassets,\n" +
                "       ceil(sp.sewithoutmi/10000) as sewithoutmi, ceil(sp.netoperatecashflow/10000) as netoperatecashflow, sp.roe,\n" +
                "   sp.oryoy, sp.grossprofityoy, sp.opyoy, sp.totalprofityoy, sp.totalassettoboy, sp.netocflowyoy, sp.enddate, sed.firstindustrycode,\n" +
                "       sed.secondindustrycode, sed.thirdindustrycode, sed.fourthindustrycode from\n" +
                "   (" +
                "on" +
                "select innercode, astockcode, companycode, provincecode, officezipcode, inducsrccode, ceil(regcapital/1000000) as regcapital from stk_companyarchives) as scp\n" +
                "   left join (\n" +
                "       SELECT c.companycode, concat_ws('|', collect_set(c.conceptcode)) as conceptcode,\n" +
                "              concat_ws('|', collect_set(l.classcode)) as classcode,\n" +
                "              concat_ws('|', collect_set(l.subclasscode)) as subclasscode\n" +
                "       FROM (\n" +
                "                SELECT conceptcode, companycode FROM stk_coconcept WHERE indistate = 1\n" +
                "            ) as c\n" +
                "                LEFT JOIN (\n" +
                "           SELECT classcode, subclasscode, conceptcode FROM stk_conceptlist\n" +
                "       ) as l\n" +
                "                          ON c.conceptcode = l.conceptcode\n" +
                "       GROUP BY c.companycode\n" +
                "   ) as sco\n" +
                "   on scp.companycode = sco.companycode\n" +
                "   left join (\n" +
                "      select t.companycode, t.pettm, t.pb  from\n" +
                "             (select companycode, pettm, pb, ROW_NUMBER() OVER (PARTITION BY companycode ORDER BY tradingdate DESC) AS row_num from stk_monthquote) as t\n" +
                "      WHERE t.row_num = 1\n" +
                "   ) as sm\n" +
                "   on sm.companycode = scp.companycode\n" +
                "   left join (\n" +
                "       select t.firstindustrycode, t.secondindustrycode, t.thirdindustrycode, t.fourthindustrycode, t.companycode  from\n" +
                "           (select firstindustrycode,\n" +
                "                   secondindustrycode, thirdindustrycode, fourthindustrycode, companycode, ROW_NUMBER() OVER (PARTITION BY companycode ORDER BY updatetime DESC) AS row_num from stk_exgindustry) as t\n" +
                "       WHERE t.row_num = 1\n" +
                "   ) as sed\n" +
                "   on sed.companycode = sm.companycode\n" +
                "   left join (\n" +
                "    select t.*  from (\n" +
                "        select operatingrevenue, grossprofit, totalprofit, totalassets, sewithoutmi, netoperatecashflow,\n" +
                "               roe, oryoy, grossprofityoy, opyoy, totalprofityoy, totalassettoboy, netocflowyoy, enddate,\n" +
                "               companycode, ROW_NUMBER() OVER (PARTITION BY companycode ORDER BY enddate DESC) AS row_num from stk_performanceletters) as t\n" +
                "       WHERE t.row_num = 1\n" +
                "   ) as sp\n" +
                "   on sp.companycode = scp.companycode");
        // 写入HDFS文件
        sqlDF.coalesce(1).write()
                .format("csv")
                .option("header", "true")
                .option("mode", "overwrite")
                .save(fileName);
    }
}
